#include "pch.h"
#include <vector>
#include "../Config/Field.h"
#include "grpc++/grpc++.h"
#include "SyncDataThread.h"
#include "SyncData.h"
#include "ImportData.h"

int SyncDataThread::row_size_ = 8096;

SyncDataThread::SyncDataThread(std::string sync_id, SyncData *sync_data, Vault *vault, std::shared_ptr<EnclaveProvider> &enclave_provider)
    : sync_id_(sync_id), sync_data_(sync_data), hana_data_(sync_data->config_->database_), vault_(vault), enclave_provider_(enclave_provider)
	, fields_(new char[row_size_ * sync_data->config_->read_count_])
{
    char address[256] = { 0 };
    sprintf_s(address, sizeof(address), "%s:%d", vault_->host_.c_str(), vault_->port_);
    auto channel = grpc::CreateChannel(address, grpc::InsecureChannelCredentials());
    if (channel.get() == nullptr)
    {
        LOG(ERROR) << "Create channel error.";
        return;
    }

    stub_ = enclave::EnclaveServer::NewStub(channel);

    request_  = stub_->SyncData(&context_, &sync_data_response_);
}


SyncDataThread::~SyncDataThread()
{
}

bool SyncDataThread::runLoop()
{
    LOG(INFO) << "Get Task";
    SyncData::Task task;
    if (sync_data_->getTask(task) == false)
    {
        LOG(INFO) << "Get Task return false";
        
        request_->WritesDone();

		auto status = request_->Finish();

		LOG(INFO) << "rquest finished, status = " << status.error_code() << ", message = " << status.error_message();

        return false;
    }
    LOG(INFO) << "Task table = " << task.table_ << ", task start = " << task.start_ << ", task start = " << task.end_;

    char sql[1024] = { 0 };
    
    sprintf_s(sql, "select * from %s where seq between %d and %d;", task.table_.c_str(), task.start_, task.end_);

    int num_rows = sync_data_->config_->read_count_;
	std::vector<Field> &columns = sync_data_->tables_columns_[task.table_];
	int num_fields = (int)columns.size();

	memset(fields_.get(), 0, row_size_ * sync_data_->config_->read_count_);
    if (hana_data_.select_sync(sql, columns, fields_.get(), num_rows) == false)
	{
		LOG(ERROR) << "select_sync error";
		return true;
	}

    int buf_len = 5120;
    char *buf = new char[buf_len];

    int values_len = 10240;
    char *values = new char[values_len];
    int encrypt_values_len = values_len * 2;
    char *encrypt_values = new char[encrypt_values_len];

    ::enclave::SyncTableRows sync_table_rows;
    sync_table_rows.set_syncid(sync_id_);
    sync_table_rows.mutable_rows()->set_pos(task.start_);
    sync_table_rows.mutable_rows()->set_length(task.end_ - task.start_ + 1);
    sync_table_rows.mutable_rows()->set_tablename(task.table_);

	LOG(INFO) << "Decrypt, Public Encrypt";

	int fields_pos = 0;

    for (int i = 0; i < num_rows; ++i)
    {
        auto row = sync_table_rows.mutable_rows()->add_rows();

        int values_pos = 0;

        memset(values, 0, values_len);
        memset(encrypt_values, 0, encrypt_values_len);

        for (int j = 0; j < num_fields; ++j)
        {
            int buf_size = columns[j].length_ == 0 ? IData::zero_size_ : columns[j].length_;

			int len = (int)*((SQLLEN *)(fields_.get() + fields_pos + buf_size));
			len = len < 0 ? 0 : len;

            if (std::find(ImportData::columns_ex_.begin(), ImportData::columns_ex_.end(), columns[j]) != ImportData::columns_ex_.end())
            {
                memcpy(values + values_pos, fields_.get() + fields_pos, len);
                row->add_value_lens(len);

                values_pos += len;
            }
            else if (columns[j].type_ == "DECIMAL" || columns[j].type_ == "INTEGER")
            {
                memcpy(values + values_pos, fields_.get() + fields_pos, len);
                row->add_value_lens(len);

                values_pos += len;
            }
            else
            {
				if (len == 0)
				{
					row->add_value_lens(0);
				}
				else
				{
					memset(buf, 0, buf_len);
					int decrypt_len = sync_data_->enclave_provider_->decrypt(fields_.get() + fields_pos, len, buf, buf_len);

					if (decrypt_len == 0)
					{
						LOG(ERROR) << "Decrypt error, values: " << std::string(fields_.get() + fields_pos, len);
						return true;
					}

					memcpy(values + values_pos, buf, decrypt_len);
					row->add_value_lens(decrypt_len);

					values_pos += decrypt_len;
				}
            }

			fields_pos += buf_size + sizeof(SQLLEN);
        }

        enclave_provider_->public_key_encrypt(values, values_pos, encrypt_values, encrypt_values_len);

        row->set_values(encrypt_values);
    }

	LOG(INFO) << "Decrypt, Public Encrypt end";

	LOG(INFO) << "sync data start";

    bool write_result = request_->Write(sync_table_rows);

	LOG(INFO) << "sync data end, wirte_result = " << write_result << "error message: " << context_.debug_error_string();

    delete[]buf;
    delete[]values;
    delete[]encrypt_values;

    if (write_result == false)
    {
        return false;
    }

    return true;
}